package es

import (
	"github.com/olivere/elastic"
	"strings"
	"fmt"
	"context"
	"time"
)

// 初始化ES，准备接收Kafka那边发过来的数据

type LogData struct{
	Topic string `json:"topic"`
	Data string `json:"data"`
}

var (
	client *elastic.Client
	ch chan *LogData
)
func Init(address string, chanSize,nums int)(err error){
	// 1.初始化连接，得到一个client
	if !strings.HasPrefix(address,"http://"){
		address = "http://"+address
	}
	client, err = elastic.NewClient(elastic.SetURL(address))
	if err != nil {
		// Handle error
		return err
	}

	fmt.Println("connect to es success")
	ch = make(chan *LogData, chanSize)
	for i :=0;i<nums;i++{
		go SendToES()
	}

	return
}


func SendToESChan(msg *LogData){
	ch <- msg
}

// 发送数据到ES
func SendToES() {

	// 链式操作
	for {
		select {
			case msg := <- ch:
				put1, err := client.Index().
					Index(msg.Topic).
					Type("xxx").
					BodyJson(msg).
					Do(context.Background())
				if err != nil {
					fmt.Println(err)
				}
				fmt.Printf("Indexed user %s to index %s, type %s\n", put1.Id, put1.Index, put1.Type)
		default:
			time.Sleep(time.Second)
		}
	}


}